package cn.witsky.rtp2ws.client.ws.netty;

import java.net.URI;
import java.util.concurrent.ThreadFactory;
import cn.witsky.rtp2ws.client.ws.ClientUtil;
import cn.witsky.rtp2ws.constant.ThreadPoolConstant;
import cn.witsky.rtp2ws.domain.Rtp;
import cn.witsky.rtp2ws.util.CapcityUtil;
import cn.witsky.rtp2ws.util.DefaultThreadFactory;
import cn.witsky.rtp2ws.util.LogUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.internal.StringUtil;

public class Netty {
	public static Bootstrap boot;
	private static int threadCount;
	private static ThreadFactory threadFactory = new DefaultThreadFactory(ThreadPoolConstant.wsClient);
	static {
		boot = new Bootstrap();
		boot.option(ChannelOption.SO_KEEPALIVE, true);
		boot.option(ChannelOption.TCP_NODELAY, true);
		boot.handler(new LoggingHandler(LogLevel.DEBUG));
		int cap=Runtime.getRuntime().availableProcessors();
		threadCount=CapcityUtil.tableSizeFor(cap);
		if(threadCount<16) {
			threadCount=16;
		}
		if (Epoll.isAvailable()) {
			boot.group(new EpollEventLoopGroup(threadCount,threadFactory));
			boot.channel(EpollSocketChannel.class);
		} else {
			boot.group(new NioEventLoopGroup(threadCount,threadFactory));
			boot.channel(NioSocketChannel.class);
		}
		boot.handler(new ChannelInitializer<SocketChannel>() {
			protected void initChannel(SocketChannel socketChannel) throws Exception {
				ChannelPipeline p = socketChannel.pipeline();
				p.addLast(new ChannelHandler[] { new HttpClientCodec(), new HttpObjectAggregator(65535) });
				p.addLast("hookedHandler", new NettyWsClientHandler());
			}
		});
	}

	public static Channel getWsClient(Rtp rtp, boolean retry) {
		Channel channel = null;
		String urlStr = ClientUtil.formatUrl(rtp, retry);
		LogUtil.info(rtp,"Netty[getWsClient:67]","task:{} ws:{}开始创建连接......", rtp.getRtpId(),urlStr);
		if (!StringUtil.isNullOrEmpty(urlStr)) {
			URI wsUri = null;
			try {
				wsUri = new URI(urlStr);
			} catch (Exception e) {
				LogUtil.error(rtp, "Netty[getWsClient:73]", "task:"+rtp.getRtpId()+"连接:"+urlStr+"出现异常:", e);
			}
			if (wsUri != null) {
				try {
					HttpHeaders httpHeaders = new DefaultHttpHeaders();
					// 进行握手
					WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(wsUri,
							WebSocketVersion.V13, (String) null, true, httpHeaders);
					channel = boot.connect(wsUri.getHost(), wsUri.getPort()).sync().channel();
					handshaker.handshake(channel);
					NettyWsClientHandler handler = (NettyWsClientHandler) channel.pipeline().get("hookedHandler");
					handler.setHandshaker(handshaker);
					handler.setRtp(rtp);		
					// 阻塞等待是否握手成功
					ChannelPromise channelPromise = handler.getHandshakeFuture().sync();
					if (channelPromise.isSuccess()) {
						LogUtil.info(rtp,"Netty[getWsClient:89]","task:{} client:{}创建成功......", rtp.getRtpId(),channel);
					} else {
						LogUtil.warn(rtp,"Netty[getWsClient:91]","task:{} client:{}创建失败......", rtp.getRtpId(),channel);
						channel.close();
						channel = null;
					}
				} catch (Exception ex) {
					LogUtil.error(rtp, "Netty[getWsClient:96]", "task:" + rtp.getRtpId() + "创建连接异常:" + ex.getMessage());
					channel.close();
					channel = null;
				}
			}
		}
		if(channel==null) {
			ClientUtil.setProxyStatus(urlStr);
		}
		return channel;
	}
}
